package com.aikg.kgrag_java.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch._types.query_dsl.RangeQuery;
import co.elastic.clients.elasticsearch.core.*;
import co.elastic.clients.json.JsonData;
import com.aikg.kgrag_java.pojo.entity.Question;
import com.aikg.kgrag_java.service.QuestionService;
import jakarta.annotation.Resource;
import org.springframework.stereotype.Service;
import co.elastic.clients.elasticsearch.core.search.Hit;

import java.io.IOException;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.stream.Collectors;

/**
 * @Author tfan
 * @Description
 * @Date 2025/1/1 21:09
 **/

@Service
public class QuestionServiceImpl implements QuestionService {

    @Resource
    private ElasticsearchClient elasticsearchClient;

    private static final String INDEX_NAME = "user_questions_v2";

    private static final double SIMILARITY_THRESHOLD = 0.7;    // 相似性阈值

    /**
     * 保存或更新一个问题
     *
     * @param question
     */
    @Override
    public void saveOrUpdateOne(Question question) {
        try {
            // 1. 检索相似问题
            SearchRequest searchRequest = SearchRequest.of(s -> s
                    .index(INDEX_NAME)
                    .query(q -> q
                            .bool(b -> b
                                    .must(m1 -> m1
                                            .match(m -> m
                                                    .field("content")
                                                    .query(question.getContent())
                                            )
                                    )
                                    .must(m2 -> m2
                                            .term(t -> t
                                                    .field("project_id.keyword")
                                                    .value(question.getProjectId())
                                            )
                                    )
                            )
                    )
                    .size(1) // 只需要检索最相似的一条记录
            );

            SearchResponse<Question> searchResponse = elasticsearchClient.search(searchRequest, Question.class);

            // 2. 检查是否有相似问题
            List<Hit<Question>> hits = searchResponse.hits().hits();
            if (!hits.isEmpty()) {
                Hit<Question> mostSimilarHit = hits.get(0);
                double similarityScore = mostSimilarHit.score();

                // 判断相似度是否高于阈值
                if (similarityScore >= SIMILARITY_THRESHOLD) {
                    String existingId = mostSimilarHit.id();
                    // 3. 更新相似问题的 frequency
                    UpdateRequest<Question, Question> updateRequest = UpdateRequest.of(u -> u
                            .index(INDEX_NAME)
                            .id(existingId)
                            .script(s -> s
                                    .inline(s1 -> s1
                                            .source("if (ctx._source.frequency == null) { ctx._source.frequency = 1; } else { ctx._source.frequency += 1; }")
                                            .lang("painless")
                                    )
                            )
                    );

                    UpdateResponse<Question> updateResponse = elasticsearchClient.update(updateRequest, Void.class);
//                    if ("updated".equals(updateResponse.result().jsonValue())) {
//                        System.out.println("Updated frequency for similar question: " + existingId);
//                    }
                    return;
                }
            }

            // 4. 如果没有相似问题，保存为新问题
            elasticsearchClient.index(i -> i
                    .index(INDEX_NAME)
                    .id(question.getId()) // 可以自定义生成 ID
                    .document(question)
            );
//            System.out.println("Saved new question with ID: " + question.getQuestionId());


        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void save(Question question) {
        try {
            elasticsearchClient.index(i -> i
                    .index(INDEX_NAME)
                    .id(question.getId()) // 可以自定义生成 ID
                    .document(question)
            );
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 批量保存
     *
     * @param questions
     */
    @Override
    public void saveBatch(List<Question> questions) {
        try {
            // 构建批量操作请求
            BulkRequest.Builder bulkRequestBuilder = new BulkRequest.Builder();

            for (Question question : questions) {
                bulkRequestBuilder.operations(op -> op
                        .index(idx -> idx
                                .index(INDEX_NAME)
                                .id(question.getId()) // 设置文档 ID，可选
                                .document(question)   // 设置文档内容
                        )
                );
            }

            // 执行批量操作
            BulkResponse response = elasticsearchClient.bulk(bulkRequestBuilder.build());

            // 处理响应
            if (response.errors()) {
                System.err.println("Errors occurred during bulk save:");
                response.items().forEach(item -> {
                    if (item.error() != null) {
                        System.err.println("Error for document ID: " + item.id() + " - " + item.error().reason());
                    }
                });
            } else {
                System.out.println("Bulk save successful!");
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 根据问题ID删除一个问题
     *
     * @param questionId
     */
    @Override
    public void deleteOne(String questionId) {
        try {
            // 执行删除操作
            DeleteResponse response = elasticsearchClient.delete(d -> d
                    .index(INDEX_NAME)
                    .id(questionId)
            );

            // 检查删除结果
            System.out.println(response.result());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 根据问题ID列表批量删除问题
     *
     * @param questionIds
     */
    @Override
    public void deleteAll(List<String> questionIds) {
        try {
            if (questionIds == null || questionIds.isEmpty()) {
                return;
            }

            BulkRequest.Builder bulkRequestBuilder = new BulkRequest.Builder();

            // 构建批量删除请求
            for (String questionId : questionIds) {
                bulkRequestBuilder.operations(op -> op
                        .delete(d -> d
                                .index("user_questions") // 索引名
                                .id(questionId)         // 指定要删除的文档 ID
                        )
                );
            }

            // 执行批量删除请求
            elasticsearchClient.bulk(bulkRequestBuilder.build());

        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public List<Question> getByKeyword(String keyword, String projectId) {
        try {
            // 构建搜索请求
            SearchResponse<Question> response = elasticsearchClient.search(s -> s
                            .index(INDEX_NAME)
                            .query(q -> q
                                    .bool(b -> b
                                            .must(m1 -> m1
                                                    .match(m -> m
                                                            .field("content")
                                                            .query(keyword)
                                                    )
                                            )
                                            .must(m2 -> m2
                                                    .term(t -> t
                                                            .field("project_id.keyword")
                                                            .value(projectId)
                                                    )
                                            )
                                    )
                            ),
                    Question.class
            );

            // 将响应中的数据转换为实体类列表
            return response.hits().hits().stream()
                    .map(Hit::source)
                    .collect(Collectors.toList());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void updateOne(Question question) {
        System.out.println(question);
        try {
            // 执行更新操作
            UpdateResponse<Question> response = elasticsearchClient.update(u -> u
                            .index(INDEX_NAME)   // 指定索引名称
                            .id(question.getId())              // 指定文档 ID
                            .doc(question), // 设置更新内容
                    Question.class
            );

            // 检查更新结果
            System.out.println(response.result());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }


    /**
     * 根据问题ID获取一个问题
     *
     * @param questionId
     * @return
     */
    @Override
    public Optional<Question> getOneById(String questionId) {
        try {
            GetResponse<Question> response = elasticsearchClient.get(g -> g
                            .index(INDEX_NAME)
                            .id(questionId),
                    Question.class
            );

            return response.found() ? Optional.of(response.source()) : Optional.empty();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 根据关键字搜索问题
     *
     * @param keyword
     * @return
     */
    @Override
    public List<Question> getByKeyword(String keyword) {
        try {
            // 构建搜索请求
            SearchResponse<Question> response = elasticsearchClient.search(s -> s
                            .index(INDEX_NAME)
                            .query(q -> q
                                    .match(m -> m
                                            .field("content")
                                            .query(keyword)
                                    )
                            ),
                    Question.class
            );

            // 将响应中的数据转换为实体类列表
            return response.hits().hits().stream()
                    .map(Hit::source)
                    .collect(Collectors.toList());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 根据多个关键字搜索问题
     * 关键字匹配越多，排名越靠前
     *
     * @param keywords           多个关键字之间用空格分隔
     * @param minimumShouldMatch 最少匹配关键字的数量
     * @param limit              最多返回的结果数量
     * @return
     */
    @Override
    public List<Question> getByKeywords(String keywords, int minimumShouldMatch, int limit, String projectId) {
        try {
            String[] keywordArray = keywords.split("\\s+");
//            System.out.println("keywords: " + Arrays.toString(keywordArray));

            SearchResponse<Question> response = elasticsearchClient.search(s -> s
                            .index(INDEX_NAME)
                            .query(q -> q
                                    .bool(b -> b
                                            .filter(f -> f
                                                    .term(t -> t
                                                            .field("project_id")
                                                            .value(projectId) // 添加 projectId 过滤条件
                                                    )
                                            )
                                            .should(Arrays.stream(keywordArray)
                                                    .map(keyword -> Query.of(qu -> qu
                                                            .match(m -> m
                                                                    .field("content")
                                                                    .query(keyword)
                                                            )
                                                    ))
                                                    .collect(Collectors.toList())
                                            )
                                            .minimumShouldMatch("" + minimumShouldMatch)
                                    )
                            )
                            .size(limit)
                            .sort(so -> so
                                    .score(sc -> sc
                                            .order(SortOrder.Desc)
                                    )
                            ),
                    Question.class
            );

            return response.hits().hits().stream()
                    .map(Hit::source)
                    .collect(Collectors.toList());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public List<Question> getTopList(int limit, String projectId) {
        try {
            // 构建搜索请求
            SearchResponse<Question> response = elasticsearchClient.search(s -> s
                            .index(INDEX_NAME)
                            .size(limit) // 限制返回的结果数量
                            .query(q -> q
                                    .term(t -> t
                                            .field("project_id")
                                            .value(projectId)
                                    )
                            )
                            .sort(so -> so
                                    .field(f -> f
                                            .field("frequency") // 按 frequency 字段排序
                                            .order(SortOrder.Desc) // 倒序排序
                                    )
                            ),
                    Question.class
            );

            // 将响应中的数据转换为实体类列表
            return response.hits().hits().stream()
                    .map(Hit::source) // 获取每个 Hit 的 source
                    .collect(Collectors.toList());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 通过脚本更新问题的访问次数
     * 使问题访问次数加 1
     * 脚本操作具有原子性，并发时保证数据的一致性
     *
     * @param questionId
     */
    @Override
    public void incrementFrequency(String questionId) {
        try {
            UpdateRequest<Question, Question> request = UpdateRequest.of(u -> u
                    .index(INDEX_NAME)
                    .id(questionId)
                    .script(s -> s
                            .inline(s1 -> s1
                                    .source("if (ctx._source.frequency == null) { ctx._source.frequency = 1; } else { ctx._source.frequency += 1; }")
                                    .lang("painless")
                            )
                    )
            );
            elasticsearchClient.update(request, Question.class);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

//    @Override
//    public List<Question> getHotQuestions(LocalDateTime startTime, LocalDateTime endTime, int limit, String projectId) {
//        try {
//            DateTimeFormatter formatter = DateTimeFormatter.ISO_LOCAL_DATE_TIME;
//
//            SearchResponse<Question> response = elasticsearchClient.search(s -> s
//                            .index(INDEX_NAME)
//                            .size(limit) // 限制返回的结果数量
//                            .query(q -> q
//                                    .bool(b -> b
//                                            .must(m -> m
//                                                    .term(t -> t
//                                                            .field("project_id")
//                                                            .value(projectId)
//                                                    )
//                                            )
//                                            .filter(f -> f
//                                                    .range(r -> r
//                                                            .field("last_viewed_date") // 时间字段，需确保与索引的实际字段名一致
//                                                            .gte(JsonData.of(startTime.format(formatter))) // 开始时间
//                                                            .lte(JsonData.of(endTime.format(formatter)))   // 结束时间
//                                                    )
//                                            )
//                                    )
//                            )
//                            .sort(so -> so
//                                    .field(f -> f
//                                            .field("frequency") // 按 frequency 字段排序
//                                            .order(SortOrder.Desc) // 倒序排序
//                                    )
//                            ),
//                    Question.class
//            );
//
//            // 处理结果
//            return response.hits().hits().stream()
//                    .map(Hit::source)
//                    .filter(Objects::nonNull)
//                    .collect(Collectors.toList());
//
//        } catch (IOException e) {
//            throw new RuntimeException("Failed to query hot questions", e);
//        }
//    }

    @Override
    public List<Question> getHotQuestions(LocalDateTime startTime, LocalDateTime endTime, int limit, String projectId) {
        try {
            DateTimeFormatter formatter = DateTimeFormatter.ISO_LOCAL_DATE_TIME;

//            String startDateStr = startTime.format(formatter);
//            String endDateStr = endTime.format(formatter);

            startTime = startTime.atZone(ZoneId.systemDefault()).withZoneSameInstant(ZoneOffset.UTC).toLocalDateTime();
            endTime = endTime.atZone(ZoneId.systemDefault()).withZoneSameInstant(ZoneOffset.UTC).toLocalDateTime();

            String startDateStr = startTime.format(formatter);
            String endDateStr = endTime.format(formatter);

            SearchResponse<Question> response = elasticsearchClient.search(s -> s
                            .index(INDEX_NAME)
                            .size(limit)
                            .query(q -> q
                                    .bool(b -> b
                                            .must(m -> m
                                                    .term(t -> t
                                                            .field("project_id.keyword") // 使用 keyword 字段
                                                            .value(projectId)
                                                    )
                                            )
                                            .filter(f -> f
                                                    .range(r -> r
                                                            .field("last_viewed_date") // 时间字段
                                                            .gte(JsonData.of(startDateStr)) // 开始时间
                                                            .lte(JsonData.of(endDateStr))   // 结束时间
                                                    )
                                            )
                                    )
                            )
                            .sort(so -> so
                                    .field(f -> f
                                            .field("frequency")
                                            .order(SortOrder.Desc)
                                            .missing("_last") // 处理缺失值
                                    )
                            ),
                    Question.class
            );

//            SearchResponse<Question> response = elasticsearchClient.search(s -> s
//                            .index(INDEX_NAME)
//                            .size(limit) // 限制返回的结果数量
//                            .query(q -> q
//                                    .bool(b -> b
//                                            .must(m -> m
//                                                    .term(t -> t
//                                                            .field("project_id")
//                                                            .value(projectId)
//                                                    )
//                                            )
//                                            .filter(f -> f
//                                                    .range(r -> r
//                                                            .field("last_viewed_date") // 时间字段，需确保与索引的实际字段名一致
//                                                            .gte(JsonData.of(startDateStr)) // 开始时间
//                                                            .lte(JsonData.of(endDateStr))   // 结束时间
//                                                    )
//                                            )
//                                    )
//                            )
//                            .sort(so -> so
//                                    .field(f -> f
//                                            .field("frequency") // 按 frequency 字段排序
//                                            .order(SortOrder.Desc) // 倒序排序
//                                    )
//                            ),
//                    Question.class
//            );

            // 处理结果
            return response.hits().hits().stream()
                    .map(Hit::source)
                    .filter(Objects::nonNull)
                    .collect(Collectors.toList());

        } catch (IOException e) {
            throw new RuntimeException("Failed to query hot questions", e);
        }
    }
}
